package com.my.study.rocketmq.order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 订单消费者（顺序消息）
 *
 * @author yyb8152
 * @since 20220504
 **/
public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.创建消费者Consumer，制定消费者组名(Push=>推的方式)
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("groupOrder");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("39.101.165.196:9876");
        //3.订阅主题Topic和Tag(*:所有的tag)
        consumer.subscribe("OrderTopic", "*");

        //4、注册消息监听器（顺序监听:MessageListenerOrderly）
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageList,
                                                       ConsumeOrderlyContext consumeOrderlyContext) {
                for (MessageExt messageExt : messageList) {
                    System.out.println("线程名称【" + Thread.currentThread().getName() + "】：" + "消费消息" + new String(messageExt.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        // 5、启动消费者
        consumer.start();

        System.out.println("消费者启动！！");
    }
}
